package com.corpgovernment.common.mq.consumer;

import com.corpgovernment.common.mq.enums.MessageBizTypeEnum;
import com.corpgovernment.common.utils.TenantUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;

/**
 * @author jhxue
 * @create 2021-12-17-16:24
 */
@Slf4j
public abstract class AbstractMultiTenantRocketOrderlyMQConsumer extends AbstractMultiTenantRocketMQConsumer {

	@Override
	protected MessageListenerConcurrently messageListener() {
		MessageBizTypeEnum bizTypeEnum = getMessageBizType();
		String tenantTopic = TenantUtils.getTenantTopic(bizTypeEnum.getTopic());
		return (var1, var2) -> {
			if (!CollectionUtils.isEmpty(var1)) {
				for (MessageExt messageExt : var1) {
					if (messageExt.getTopic().equals(tenantTopic)) {
						try {
							this.consume(messageExt, var2);
						} catch (Exception e) {
							log.warn("消费topic:{},失败，消息体：{}，打印异常：{}", tenantTopic, messageExt, e);
							return ConsumeConcurrentlyStatus.RECONSUME_LATER;
						}
					} else {
						// 如果没有return success, consumer会重新消费该消息, 直到return success
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
				}
			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		};
	}

}
